use crate::{
    base,
    common::{
        constant,
        cs,
        orders,
        config::{
            InnerSerialConfig,
            SerialConfig,
        },
    },
    serial::{
        basic,
        ConnectSerialConfig,
        SerialReceiver,
        SerialSender,
        enhanced,
    },
    modbus::basic::ModbusRtuResult,
    database::dao,
};
use std::time::Duration;
use std::collections::HashMap;
use tokio::sync::mpsc;
use log::{debug, info, warn};
use scheduled_thread_pool::ScheduledThreadPool;

/// 初始化串口
pub async fn init_serial_port(
    scheduled: &ScheduledThreadPool,
    truncate_device: bool,
    serial_config: &SerialConfig,
)
    -> cs::Result<()>
{
    let ps = basic::get_ports()?;
    println!(" - starting init serial ports : `{:?}`", ps);

    if truncate_device{
        dao::delete_all_device().await?;
    }

    // 需要重置的数据库
    let mut devices = dao::select_all_device().await?;

    for (path, inner_config) in &serial_config.mapping {
        // 验证是否有效串口 无效则跳过
        if !ps.contains(path) {
            warn!(" - serial port `{}` is absence", path);
            continue;
        };
        debug!("path: `{}`, config: `{:?}`",path,inner_config);
        // 解析出所有命令 以及查询数据库设备id 关联到执行逻辑
        // 需要循环发送的命令
        let mut orders: Vec<Box<[u8]>> = vec![];
        // 初始化时需要发送的命令
        let mut init_orders: Vec<Box<[u8]>> = vec![];
        // 设备与device_id的关联 如果在数据库中不存在会新建
        let mut device_map = HashMap::new();
        // 初始化天气相关
        for weather_address in &inner_config.weather_address {
            let order = orders::weather_get_order(*weather_address as u8);
            orders.push(order.to_vec().into_boxed_slice());
            let order = orders::weather_init_order(*weather_address as u8);
            init_orders.push(order.to_vec().into_boxed_slice());
            let (new, id) = select_new_serial_device_id(
                path, *weather_address, constant::DEVICE_WEATHER).await?;
            if !new && truncate_device { exclude_valid_device_id(&mut devices, id); }
            device_map.insert(*weather_address, id);
        }
        // 初始化土壤相关
        for soil_address in &inner_config.soil_address {
            let order = orders::soil_get_order(*soil_address as u8);
            orders.push(order.to_vec().into_boxed_slice());
            let (new, id) = select_new_serial_device_id(
                path, *soil_address, constant::DEVICE_SOIL).await?;
            if !new && truncate_device { exclude_valid_device_id(&mut devices, id); }
            device_map.insert(*soil_address, id);
        }
        // 初始化雨雪相关
        for sleet_address in &inner_config.sleet_address {
            let order = orders::sleet_get_order(*sleet_address as u8);
            orders.push(order.to_vec().into_boxed_slice());
            let (new, id) = select_new_serial_device_id(
                path, *sleet_address, constant::DEVICE_SLEET).await?;
            if !new && truncate_device { exclude_valid_device_id(&mut devices, id); }
            device_map.insert(*sleet_address, id);
        }
        // 初始化开关相关与集成开关相关
        for switch_address in &inner_config.switch_address {
            let order = orders::switch_get_order(*switch_address as u8);
            orders.push(order.to_vec().into_boxed_slice());
            let (new, id) = select_new_serial_device_id(
                path, *switch_address, constant::DEVICE_SWITCH).await?;
            if !new && truncate_device { exclude_valid_device_id(&mut devices, id); }
            device_map.insert(*switch_address, id);
        }
        debug!("0");
        let mut custom_order = false;
        for address in &inner_config.paralleling_address {
            let order = match orders::paralleling_get_order(address) {
                Some(o) => { o }
                None => {
                    warn!("invalid address : `{}`",address);
                    continue;
                }
            };
            orders.push(order.to_vec().into_boxed_slice());
            let (new, id) = select_new_serial_device_id(
                path, 0, constant::DEVICE_PARALLELING_SWITCH).await?;
            if !new && truncate_device { exclude_valid_device_id(&mut devices, id); }
            custom_order = true;
        }
        // 开启串口并执行线程
        let (s, r) = open_serial_async(
            path,
            ConnectSerialConfig {
                baud_rate: inner_config.baud_rate,
                parity: inner_config.parity,
                data_bits: inner_config.data_bits,
                stop_bits: inner_config.stop_bits,
                timeout_millis: inner_config.serial_timeout,
                buffer_size: 16,
                read_delay_millis: inner_config.read_delay_millis,
                reread_times: inner_config.reread_times,
            },
        ).await?;
        // 存储sender的复制
        base::serial_sender().insert(path.to_string(), s.clone());
        // 执行sender线程 按照配置定时发送并且监听返回信息
        exec_sender(scheduled, init_orders, orders, inner_config.rate_delay_sec, inner_config.exec_delay_sec, path).await;
        if !custom_order {
            tokio::spawn(parse_listen(r, path.to_string(), device_map));
        } else {
            let device_id = dao::select_device_id(path, 0).await?;
            tokio::spawn(parse_listen_custom(r, path.to_string(), device_id));
        }
    }
    if truncate_device && !devices.is_empty() {
        debug!("remove device : `{:?}`",devices);
        for (device_id, _, _, _) in devices {
            dao::delete_device_by_id(device_id).await?;
        }
    }
    println!(" - serial port init complete");
    Ok(())
}

/// 开启串口
pub async fn open_serial_async<'a>(
    path: &str,
    csc: ConnectSerialConfig,
)
    -> cs::Result<(SerialSender, SerialReceiver)>
{
    let (sender, r) = mpsc::channel(1);
    match enhanced::connect_serial_port_async(path, r, csc).await {
        Ok(receiver) => {
            info!("async open serial success for : {:?}",path);
            Ok((sender, receiver))
        }
        Err(e) => {
            info!("async open serial failed for : {:?}",path);
            Err(e.into())
        }
    }
}

/// 执行发送线程
async fn exec_sender(
    scheduled: &ScheduledThreadPool,
    init_orders: Vec<Box<[u8]>>,
    orders: Vec<Box<[u8]>>,
    rate_delay_sec: u64,
    exec_delay_sec: u64,
    path: &str,
)
{
    send_order(&init_orders, path, exec_delay_sec).await;
    let p = path.to_string();
    scheduled.execute_at_fixed_rate(
        Duration::new(5, 0), Duration::from_secs(rate_delay_sec),
        move || { futures::executor::block_on(send_order(&orders, &p, exec_delay_sec)) },
    );
    info!(" - starting exec sender scheduled for : `{}`",path);
}

/// 发送命令
async fn send_order(orders: &[Box<[u8]>], path: &str, exec_delay_sec: u64)
{
    for order in orders {
        let sender = base::serial_sender().get(path).unwrap().clone();
        sender.send(order.clone()).await.unwrap();
        std::thread::sleep(Duration::from_secs(exec_delay_sec));
    }
}

/// 监听线程
async fn parse_listen(mut receiver: SerialReceiver, path: String, device_map: HashMap<u32, i32>)
{
    debug!(" - starting listen `{}`",path);
    loop {
        match receiver.recv().await.ok_or("read err in listen serial port") {
            Ok(x) => {
                if x.0 == 0 { continue; }
                let receive = &x.1[..x.0];
                info!("receive <- size:{} ~ `{:?}`",x.0,receive);
                match ModbusRtuResult::from(receive) {
                    Ok(mr) => {
                        let config = base::serial_config().mapping.get(&path).unwrap();
                        let device_id = device_map.get(&(mr.address as u32)).unwrap();
                        split_exec(config, &mr, &path, *device_id).await;
                    }
                    Err(e) => { info!("serial port `{}` read return is not modbus : `{}`",path,e) }
                };
            }
            Err(e) => { warn!("receiver err:`{}`",e); }
        };
    }
}

/// 自定义监听线程
async fn parse_listen_custom(mut receiver: SerialReceiver, path: String, device_id: i32)
{
    debug!(" - starting listen `{}`",path);
    loop {
        match receiver.recv().await.ok_or("read err in listen serial port") {
            Ok(x) => {
                let receive = &x.1[..x.0];
                info!("receive <- size:{} ~ `{:?}`",x.0,receive);
                if receive.len() != 10
                    || receive.get(8).unwrap().ne(&0x00)
                    || receive.get(9).unwrap().ne(&0x0f)
                {
                    info!("serial port `{}` read return is not custom : `{:?}`",path,receive);
                    continue;
                }
                let open1 = 1.eq(receive.get(5).unwrap());
                let open2 = 1.eq(receive.get(6).unwrap());
                let open3 = 1.eq(receive.get(7).unwrap());
                if let Err(e) = dao::insert_paralleling(device_id, open1, open2, open3).await {
                    warn!("insert custom err : {}",e);
                };
            }
            Err(e) => { warn!("receiver err:`{}`",e); }
        };
    }
}

/// 分流执行
async fn split_exec(config: &InnerSerialConfig, mr: &ModbusRtuResult, path: &str, device_id: i32)
{
    if config.sleet_address.contains(&(mr.address as u32)) {
        match orders::analysis_sleet(mr) {
            Ok(r) => {
                info!("analysis <---- {} : {:?}",path, r);
                let result = dao::insert_sleet(device_id, &r).await;
                debug!("insert result : {:?}",result);
            }
            Err(e) => { warn!("parse order error : `{}`",e) }
        };
    } else if config.soil_address.contains(&(mr.address as u32)) {
        match orders::analysis_soil(mr) {
            Ok(r) => {
                info!("analysis <---- {} : {:?}",path, r);
                let result = dao::insert_soil(device_id, &r).await;
                debug!("insert result : {:?}",result);
            }
            Err(e) => { warn!("parse order error : `{}`",e) }
        };
    } else if config.weather_address.contains(&(mr.address as u32)) {
        match orders::analysis_weather(mr) {
            Ok(r) => {
                info!("analysis <---- {} : {:?}",path, r);
                let result = dao::insert_weather(device_id, &r).await;
                debug!("insert result : {:?}",result);
            }
            Err(e) => { warn!("parse order error : `{}`",e) }
        };
    } else if config.switch_address.contains(&(mr.address as u32)) {
        match orders::analysis_switch(mr) {
            Ok(s) => {
                info!("analysis <---- {} : {:?}",path, s);
                let result = dao::insert_switch(device_id, s).await;
                debug!("insert result : {:?}",result);
            }
            Err(e) => { warn!("parse order error : `{}`",e) }
        }
    }
}

/// 获取设备id 如果没有则新增返回
async fn select_new_serial_device_id(
    port: &str,
    address: u32,
    device_type: u32,
)
    -> cs::Result<(bool, i32)>
{
    match dao::select_device_id(port, address).await {
        Ok(i) => { Ok((false, i)) }
        Err(_) => {
            let i = dao::insert_device(device_type, port, address).await?;
            Ok((true, i))
        }
    }
}

/// 排除有效的设备id
pub fn exclude_valid_device_id(v: &mut Vec<(i32, String, i32, i32)>, id: i32) {
    let mut index = -1;
    for i in 0..v.len() {
        let (device_id, _, _, _) = v.get(i).unwrap();
        if id.eq(device_id) { index = i as i32; };
    }
    if index != -1 {
        debug!("exclude device: `{}` and index of `{:?}` is `{}`",id,v,index);
        v.remove(index as usize);
    }
}

#[cfg(test)]
mod tests {
    use crate::server::serial_server;

    #[tokio::test]
    async fn test() {
        let (new, id) = serial_server::select_new_serial_device_id("COM2", 1, 1).await.unwrap();
        println!("{:?}", id);
    }

    #[test]
    fn test2() {
        let mut devices = vec![];
        devices.push((9, "".to_string(), 0, 0));
        devices.push((10, "".to_string(), 0, 0));
        devices.push((11, "".to_string(), 0, 0));
        devices.push((12, "".to_string(), 0, 0));
        serial_server::exclude_valid_device_id(&mut devices, 9);
        println!("{:?}", devices);
    }
}